#include <sys/epoll.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>

#include "iomanager.h"
#include "log.h"
#include "macro.h"

static johnsonli::Logger::ptr g_logger = LOG_NAME("system");

namespace johnsonli
{

    IOManager::FdContext::EventContext& IOManager::FdContext::getEventContext(Event event)
    {
        switch(event) {
        case IOManager::Event::READ:
            return read;
        case IOManager::Event::WRITE:
            return write;
        default:
            DO_ASSERT2(false, "getContext");

        }
        throw std::invalid_argument("getContext invalid event");
    }

    void IOManager::FdContext::resetEventContext(IOManager::FdContext::EventContext& ctx)
    {
        ctx.scheduler = nullptr;
        ctx.fiber.reset();
        ctx.cb = nullptr;
    }

    void IOManager::FdContext::triggerEvent(Event event)
    {
        DO_ASSERT(events & event);
        events = (Event)(events & ~event);

        EventContext& ctx = getEventContext(event);
        if(ctx.cb) {
            ctx.scheduler->schedule(&ctx.cb);
        } else {
            ctx.scheduler->schedule(&ctx.fiber);
        }
        ctx.scheduler = nullptr;

        return;
    }

    IOManager::IOManager(size_t threads, bool use_caller, const std::string& name)
        :Scheduler(threads, use_caller, name)
    {

        //1. 初始化Scheduler
        //2. 创建m_epfd
        //3. 监听m_tickleFds[0]读事件
        //4. 初始化m_fdContexts
        //5. 运行调度器

        m_epfd = epoll_create(5000);
        DO_ASSERT(m_epfd > 0);
         
        //给m_tickleFds[0]注册读事件，当加入任务时，可以往m_tickleFds[1]写，保证程序不会被阻塞，从而监听到新加入的任务
        int rt = pipe(m_tickleFds);
        DO_ASSERT(!rt);

        epoll_event event;
        memset(&event, 0, sizeof(epoll_event));
        event.events = EPOLLIN | EPOLLET; //监听读事件，边沿触发（一次触发后，之后不再触发，一般设置为非阻塞轮询）
        event.data.fd = m_tickleFds[0];

        //非阻塞轮询
        rt = fcntl(m_tickleFds[0], F_SETFL, O_NONBLOCK);
        DO_ASSERT(!rt);

        rt = epoll_ctl(m_epfd, EPOLL_CTL_ADD, m_tickleFds[0], &event);
        DO_ASSERT(!rt);

        contextResize(32);
        
        start();    //初始化好后就开始Scheduler的start
        
    }

    IOManager::~IOManager()
    {
        stop();

        close(m_epfd);
        close(m_tickleFds[0]);
        close(m_tickleFds[1]);

        for(size_t i = 0; i < m_fdContexts.size(); ++i)
        {
            if(m_fdContexts[i])
            {
                delete m_fdContexts[i];
            }
        }

    }

    void IOManager::contextResize(size_t size)
    {
        m_fdContexts.resize(size);

        for(size_t i = 0; i < m_fdContexts.size(); ++i)
        {
            //非null才创建，之前创建过的不会变化
            if(!m_fdContexts[i])
            {
                m_fdContexts[i] = new FdContext;
                // m_fdContexts[i]->fd = i;
            }
        }
    }

    int IOManager::addEvent(int fd, Event event, std::function<void()> cb) 
    {

        //1. 从m_fdContexts中拿到对应的fd: fd_ctx
        //2. 修改fd_ctx
        //3. 添加到m_epfd

        FdContext* fd_ctx = nullptr;
        MutexType::ReadLock lock(m_mutex);
        //从m_fdContexts中拿到对应的fd
        if((int)m_fdContexts.size() > fd)
        {
            fd_ctx = m_fdContexts[fd];  //下标志就是对应的fd
            lock.unlock();
        }
        else
        {
            lock.unlock();
            MutexType::WriteLock lock2(m_mutex);
            contextResize(fd * 1.5);
            fd_ctx = m_fdContexts[fd];
        }

        //修改fd
        FdContext::MutexType::Lock lock2(fd_ctx->mutex);
        fd_ctx->fd = fd;
        if(fd_ctx->events & event)  //如果fd_ctx上已经有这个事件了，出错
        {
            LOG_ERROR(g_logger) << "addEvent assert fd=" << fd
            << " event=" << (EPOLL_EVENTS)event
            << " fd_ctx.event=" << (EPOLL_EVENTS)fd_ctx->events;
            DO_ASSERT(!(fd_ctx->events & event));
        }
        Event old_event = fd_ctx->events;
        fd_ctx->events = (Event)(fd_ctx->events | event);
        
        FdContext::EventContext& event_ctx = fd_ctx->getEventContext(event);
        DO_ASSERT(!event_ctx.scheduler
            && !event_ctx.fiber
            && !event_ctx.cb);

        event_ctx.scheduler = Scheduler::GetThis();
        if(cb)
        {
            event_ctx.cb.swap(cb);
        }
        else
        {
            event_ctx.fiber = Fiber::GetThis();
            DO_ASSERT2(event_ctx.fiber->getState() == Fiber::EXEC
                        ,"state=" << event_ctx.fiber->getState());
        }

        //添加到m_epfd
        int op = old_event ? EPOLL_CTL_MOD : EPOLL_CTL_ADD;    //fd_ctx之前有，添加；否则修改
        epoll_event epevent;
        epevent.events = EPOLLET | fd_ctx->events;
        epevent.data.ptr = fd_ctx;

        int rt = epoll_ctl(m_epfd, op, fd, &epevent);
        if(rt) {
            LOG_ERROR(g_logger) << "epoll_ctl(" << m_epfd << ", "
                << op << ", " 
                << fd << ", " << (EPOLL_EVENTS)epevent.events << "):"
                << rt << " (" << errno << ") (" << strerror(errno) << ") fd_ctx->events="
                << (EPOLL_EVENTS)fd_ctx->events;
            return -1;
        }

        //添加了一个事件
        ++m_pendingEventCount;

        return 0;
    }

    bool IOManager::delEvent(int fd, Event event) 
    {
        //1. 从m_fdContexts中拿到对应的fd: fd_ctx
        //2. 修改fd_ctx
        //3. 从m_epfd删除

        MutexType::ReadLock lock(m_mutex);

        //1. 从m_fdContexts中拿到对应的fd: fd_ctx
        if((int)m_fdContexts.size() <= fd)
        {
            return false;
        }
        FdContext* fd_ctx = m_fdContexts[fd];
        lock.unlock();

        //2. 修改fd_ctx
        if(!(fd_ctx->events & event))    //fd_ctx中没有该事件
        {
            return false;
        }
       // Event old_event = fd_ctx->events;
        fd_ctx->events = (Event)(fd_ctx->events & ~event);
        FdContext::EventContext& event_ctx = fd_ctx->getEventContext(event);
        fd_ctx->resetEventContext(event_ctx);

        //3. 从m_epfd删除
        int op = fd_ctx->events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL; //减完event有，修改；没有，删除
        epoll_event epevent;
        epevent.events = EPOLLET | fd_ctx->events;
        epevent.data.ptr = fd_ctx;

        int rt = epoll_ctl(m_epfd, op, fd, &epevent);
        if(rt) {
            LOG_ERROR(g_logger) << "epoll_ctl(" << m_epfd << ", "
                << op << ", " 
                << fd << ", " << (EPOLL_EVENTS)epevent.events << "):"
                << rt << " (" << errno << ") (" << strerror(errno) << ")";
            return false;
        }

        //删除了一个事件
        --m_pendingEventCount;
        return true;
    }

    bool IOManager::cancelEvent(int fd, Event event) 
    {
        //1. 从m_fdContexts中拿到对应的fd: fd_ctx
        //2. 从m_epfd上删除
        //3. 触发事件

        MutexType::ReadLock lock(m_mutex);

        //1. 从m_fdContexts中拿到对应的fd: fd_ctx
        if((int)m_fdContexts.size() <= fd)
        {
            return false;
        }
        FdContext* fd_ctx = m_fdContexts[fd];
        lock.unlock();

        //2. 从m_epfd上删除
        FdContext::MutexType::Lock lock2(fd_ctx->mutex);
        if(!(fd_ctx->events & event)) {
            return false;
        }
        Event new_events = (Event)(fd_ctx->events & ~event);

        int op = new_events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
        epoll_event epevent;
        epevent.events = EPOLLET | new_events;
        epevent.data.ptr = fd_ctx;

        int rt = epoll_ctl(m_epfd, op, fd, &epevent);
        if(rt) {
            LOG_ERROR(g_logger) << "epoll_ctl(" << m_epfd << ", "
                << op << ", " << fd << ", " << (EPOLL_EVENTS)epevent.events << "):"
                << rt << " (" << errno << ") (" << strerror(errno) << ")";
            return false;
        }

        //3. 触发事件
        fd_ctx->triggerEvent(event);
        --m_pendingEventCount;
        return true;
        
    }

    bool IOManager::cancelAll(int fd) 
    {
        //1. 从m_fdContexts中拿到对应的fd: fd_ctx
        MutexType::ReadLock lock(m_mutex);
        if((int)m_fdContexts.size() <= fd) {
            return false;
        }
        FdContext* fd_ctx = m_fdContexts[fd];
        lock.unlock();

        //2. 从m_epfd上删除所有事件
        FdContext::MutexType::Lock lock2(fd_ctx->mutex);
        if(!fd_ctx->events) {   //没有任何事件
            return false;
        }

        int op = EPOLL_CTL_DEL;
        epoll_event epevent;
        epevent.events = 0;
        epevent.data.ptr = fd_ctx;

        int rt = epoll_ctl(m_epfd, op, fd, &epevent);
        if(rt) {
            LOG_ERROR(g_logger) << "epoll_ctl(" << m_epfd << ", "
                << op << ", " << fd << ", " << (EPOLL_EVENTS)epevent.events << "):"
                << rt << " (" << errno << ") (" << strerror(errno) << ")";
            return false;
        }

        //3. 触发事件
        if(fd_ctx->events & READ) {
            fd_ctx->triggerEvent(READ);
            --m_pendingEventCount;
        }
        if(fd_ctx->events & WRITE) {
            fd_ctx->triggerEvent(WRITE);
            --m_pendingEventCount;
        }
        
        //所有事件已经删除完毕
        DO_ASSERT(fd_ctx->events == 0);
        return true;
    }

    IOManager* IOManager::GetThis()
    {
        return dynamic_cast<IOManager*>(Scheduler::GetThis());
    }

    void IOManager::tickle()
    {
        //LOG_INFO(g_logger) << "iom tickle";
        //没有空闲的线程，就不用唤醒
        if(!hasIdleThreads())
        {
            return;
        }

        int rt = write(m_tickleFds[1], "T", 1);
        DO_ASSERT(rt == 1);
    }

    bool IOManager::stopping()
    {
        return Scheduler::stopping() && m_pendingEventCount == 0;
    }

    void IOManager::idle()
    {
//        LOG_DEBUG(g_logger) << "idle";
        
        // 协程栈中，不能开的太大的空间，所以用指针
        // epoll_wait最多监听256个
        const uint64_t MAX_EVNETS = 256;
        epoll_event* events = new epoll_event[MAX_EVNETS]();

        std::shared_ptr<epoll_event> shared_events(events, [](epoll_event* ptr){
            delete[] ptr;
        });


        while(true)
        {
           // LOG_INFO(g_logger) << "idle in";
            uint64_t next_timeout = 0;
            next_timeout = getNextTimer();
            //结束
            if(stopping())
            {
                // LOG_INFO(g_logger) << "next_timeout = " << next_timeout;
                //没有定时器，就结束
                if(next_timeout == ~0ull)
                {
                    LOG_DEBUG(g_logger) << "name=" << getName() << " idle stopping exit";
                    break;
                }
               
            }


            int rt = 0;

            //监听事件，等待注册的事件发生
            do
            {
                static const int MAX_TIMEOUT = 5000;

                if(next_timeout != ~0ull)
                {
                    //有定时器，就选择最小的那个
                    next_timeout = (int)next_timeout > MAX_TIMEOUT
                                ? MAX_TIMEOUT : next_timeout;
                }
                else
                {
                    next_timeout = MAX_TIMEOUT;
                }



                //优先考虑最先执行的那个定时器
                rt = epoll_wait(m_epfd, events, 64, (int)next_timeout); //一般情况下，next_timeout=0，因为如果有阻塞时长，将会在这一直等待，如果此时有新任务来了，就无法马上新任务了

                if(rt < 0 && errno == EINTR)    //中断引发的错误，继续
                {}
                else
                {
                    break;  //有事件触发，跳出
                }
            } while (true);


            //定时器到期的任务加入到任务协程
            std::vector<std::function<void()>> cbs;
            listExpiredCb(cbs);
            if(!cbs.empty()) {
//                LOG_INFO(g_logger) << "!cbs.empty()[0]" ;
                schedule(cbs.begin(), cbs.end());
                cbs.clear();
            }

            

            for(int i = 0; i < rt; ++i)
            {
                epoll_event& event = events[i];

                //如果这个是m_tickleFds[0], 说明tickle()发送了一个数据，触发了m_tickleFds[0]读事件
                if(event.data.fd == m_tickleFds[0])
                {
                    uint8_t dummy[256];
                    //m_tickleFds[0]：fd 4
                    //LOG_INFO(g_logger) << "m_tickleFds[0] : " << m_tickleFds[0];
                    while(read(m_tickleFds[0], dummy, sizeof(dummy)) > 0);  //非阻塞轮询
                    continue;  
                }

                //处理自己注册的事件
                FdContext* fd_ctx = (FdContext*)event.data.ptr;
                FdContext::MutexType::Lock lock(fd_ctx->mutex);

                if(event.events & (EPOLLERR | EPOLLHUP))
                {
                    event.events |= (EPOLLIN | EPOLLHUP) & fd_ctx->events;
                }

                //查看触发了哪些事件
                int real_events = NONE;
                if(event.events & EPOLLIN) {
                    real_events |= READ;
                }
                if(event.events & EPOLLOUT) {
                    real_events |= WRITE;
                }

                //fd_ctx中没有注册这些事件，不触发
                if((fd_ctx->events & real_events) == NONE) {
                    continue;
                }

                //减去触发了的事件
                int left_events = (fd_ctx->events & ~real_events);
                int op = left_events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
                event.events = EPOLLET | left_events;

                int rt2 = epoll_ctl(m_epfd, op, fd_ctx->fd, &event);
                if(rt2) {
                    LOG_ERROR(g_logger) << "epoll_ctl(" << m_epfd << ", "
                        << op << ", " << fd_ctx->fd << ", " << (EPOLL_EVENTS)event.events << "):"
                        << rt2 << " (" << errno << ") (" << strerror(errno) << ")";
                    continue;
                }


                //执行事件处理函数
                if(real_events & READ) {
                    fd_ctx->triggerEvent(READ);
                    --m_pendingEventCount;
                }
                if(real_events & WRITE) {
                    fd_ctx->triggerEvent(WRITE);
                    --m_pendingEventCount;
                }

            }

            //没处理完一个事件后，都要返回到调度协程，看看有没有协程任务需要处理
            johnsonli::Fiber::YieldToHoldBySwap();
        }

            
        
        
    }


    void IOManager::onTimerInsertedAtFront() 
    {
        tickle(); //当有新的定时器插入在头部，需要唤醒epoll_wait，重新设置定时
    }
}
